/**
 * 
 */
package mapreduce4j;

import java.nio.ByteBuffer;
import java.util.regex.Pattern;

import org.apache.hadoop.io.Text;

/**
 * Reads a KVP from a TAB file 
 * @author tim
 */
public class KVPRecordReader extends RecordReader<Text, Text> {
	protected ByteBuffer buffer;
	protected Text k;
	protected Text v;
	public static final char lineTerminator = '\n';
	protected static Pattern tab = Pattern.compile("\t");

	
	@Override
	public Text getCurrentKey() {
		return k;
	}

	@Override
	public Text getCurrentValue() {
		return v;
	}

	@Override
	public void initialize(ByteBuffer input) {
		// make a read only copy just in case
		buffer = input.asReadOnlyBuffer();
	}

	@Override
	public boolean nextKeyValue() {
		StringBuffer sb = new StringBuffer();
		while (buffer.hasRemaining()) {
			char value = (char)buffer.get();
			if (lineTerminator == value || !buffer.hasRemaining()) {
				if (!buffer.hasRemaining()) {
					sb.append(value);
				}
				String line = sb.toString();
				String parts[] = tab.split(line);

				k = new Text(parts[0]);
				v = new Text(parts[1]);
				return true;
			} else {
				sb.append(value);
			}
		}
		return false;
	}
}
